package com.wsjj.gmall.util;

import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import com.wsjj.gmall.constant.Constant;
import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.connect.runtime.ConnectorConfig;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Properties;


//连接kafka的Source

/*
知识点： 如何保证精确一次性：
    kafkaSource：从kafka主题中读取数据
                通过手动维护便宜量，保证消费的精确一次
    kafkaSink：向kafka主题中写入数据，也可以保证写入的精确一次，需要做如下操作
                开启检查点
                消费端精确一次性配置     生产环境必须写   .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
                事务前缀id，必须唯一     生产环境必须写   .setTransactionalIdPrefix(TransactionalIdPrefix)
                设置事务的超时时间：      检查点超时时间 <=  事务的超时时间  <= 事务最大超时时间(默认15分钟)   生产环境必须写     .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*10000+"")
                消费端开启精确一次性：        生产环境必须写    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)    意思是提交在kafka的数据才能读出来，预提交的数据不能读出来，才能保证  精确一次性，开发环境一定要写






 */


public class FlinkSourceUtil {

    /**
     * 连接kafka的source，从kafka读取数据
     * @param topic     kafka的topic
     * @param groupId   kafka的消费者id
     * @return
     */
    public static KafkaSource<String> getKafkaSource(String topic,String groupId){

       return KafkaSource.<String>builder()
               .setBootstrapServers(Constant.KAFKA_BROKERS)
               .setTopics(topic)
               .setGroupId(groupId)

//               注意 ：设置kafka   读已提交   （意思是提交在kafka的数据才能读出来，预提交的数据不能读出来，才能保证  精确一次性，开发环境一定要写，测试可以不写）
//               .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")

               .setStartingOffsets(OffsetsInitializer.latest())
//               注意：如果使用flink提供的SimpleStringSchema对String类型的消息进行反序列化，如果消息为空，会报错
               .setValueOnlyDeserializer(new DeserializationSchema<String>() {
                   @Override
                   public String deserialize(byte[] message) throws IOException {
                       if(message != null){
                           return new String(message, StandardCharsets.UTF_8);
                       }
                       return null;
                   }

                   @Override
                   public boolean isEndOfStream(String nextElement) {
                       return false;
                   }

                   @Override
                   public TypeInformation<String> getProducedType() {
                       return Types.STRING;
                   }
               })
               .build();

    }

    /**
     * 使用flinkCDC在mysql中读取数据
     * @param database  数据库名称
     * @param tableName     表名称
     * @return
     */
    public static MySqlSource<String> getMySqlSource(String database,String tableName){
        Properties properties = new Properties();
        properties.setProperty("useSSL","false");
        properties.setProperty("allowPublicKeyRetrieval","true");


        MySqlSource<String> gmallConfig = MySqlSource.<String>builder()
                .hostname(Constant.MYSQL_HOST)
                .port(Constant.MYSQL_Port)
                .databaseList(database)
                .tableList(database+"."+tableName)
                .username(Constant.MYSQL_USER_NAME)
                .password(Constant.MYSQL_PASSWORD)
                .jdbcProperties(properties)
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        return gmallConfig;

    }


}
